package com.laycoding.consumer.service;

import com.laycoding.common.entity.RpcMessage;
import com.laycoding.common.util.LoadBlance;
import com.laycoding.common.util.ZkUtils;
import com.laycoding.consumer.handler.RpcProxyHandler;
import com.laycoding.consumer.handler.ZkConsumerHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import org.apache.curator.framework.CuratorFramework;

import java.util.List;

public class RpcServiceImpl implements RpcService{
    @Override
    public Object sendMessage(String url, Integer port, RpcMessage message) {
        final RpcProxyHandler consumerHandler = new RpcProxyHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                            //自定义协议编码器
                            pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                            //对象参数类型编码器
                            pipeline.addLast("encoder", new ObjectEncoder());
                            //对象参数类型解码器
                            pipeline.addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
                            pipeline.addLast("handler", consumerHandler);
                        }
                    });
              //CuratorFramework client = ZkUtils.getClient();
            String rpcservice = new ZkConsumerHandler().getUrl(ZkUtils.getClient(), "/order", "rpcservice");
            String[] split = rpcservice.split(":");
            ChannelFuture future = bootstrap.connect(split[0], Integer.valueOf(split[1])).sync();
            future.channel().writeAndFlush(message).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
        return consumerHandler.getResponse();
    }
}
